package com.dukk.sedona;

import org.apache.sedona.core.formatMapper.WktReader;
import org.apache.sedona.core.spatialRDD.SpatialRDD;
import org.apache.sedona.sql.utils.Adapter;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import org.locationtech.jts.geom.Geometry;

import java.io.IOException;

public class AggregateFunctionTest extends TestBase{

    @BeforeClass
    public static void onceExecutedBeforeAll() {
        initialize(AggregateFunctionTest.class.getName());
    }

    @AfterClass
    public static void tearDown() throws IOException {
        sc.stop();
    }


    @Test
    public void ST_Envelope_Aggr() {

        String path = AggregateFunctionTest.class.getClassLoader().getResource("wkt.wkt").getPath();

        SpatialRDD<Geometry> wktRDD = WktReader.readToGeometryRDD(sc, path, 0, true, false);
        Dataset<Row> dataset = Adapter.toDf(wktRDD, sparkSession);
        dataset.show();
        dataset.printSchema();

        dataset.createOrReplaceTempView("wkt_table");

        Dataset<Row> aggrRows = sqlContext.sql("SELECT ST_Envelope_Aggr(geometry) as polygon FROM wkt_table");
        aggrRows.show();
        System.out.println(((Geometry)aggrRows.first().getAs("polygon")).toText());


    }


    @Test
    public void ST_Intersection_Aggr() {

        String path = AggregateFunctionTest.class.getClassLoader().getResource("wkt.wkt").getPath();

        SpatialRDD<Geometry> wktRDD = WktReader.readToGeometryRDD(sc, path, 0, true, false);
        Dataset<Row> dataset = Adapter.toDf(wktRDD, sparkSession);
        dataset.show();
        dataset.printSchema();

        dataset.createOrReplaceTempView("inter_aggr_table");

        Dataset<Row> aggrRows = sqlContext.sql("SELECT ST_Intersection_Aggr(geometry) as polygon FROM inter_aggr_table");
        aggrRows.show();
        System.out.println(((Geometry)aggrRows.first().getAs("polygon")).toText());


    }

    @Test
    public void ST_Union_Aggr() {

        String path = AggregateFunctionTest.class.getClassLoader().getResource("wkt2.wkt").getPath();

        SpatialRDD<Geometry> wktRDD = WktReader.readToGeometryRDD(sc, path, 0, true, false);
        Dataset<Row> dataset = Adapter.toDf(wktRDD, sparkSession);
        dataset.show();
        dataset.printSchema();

        dataset.createOrReplaceTempView("wkt_table");

        Dataset<Row> aggrRows = sqlContext.sql("SELECT ST_Union_Aggr(geometry) as polygon FROM wkt_table");
        aggrRows.show();
        System.out.println(((Geometry)aggrRows.first().getAs("polygon")).toText());


    }
}
